package avicit.bdp.dds.server.master.runner;


import avicit.bdp.common.utils.SpringApplicationContext;
import avicit.bdp.dds.dispatch.enums.ExecutionStatus;
import avicit.bdp.dds.dispatch.enums.TaskTimeoutStrategy;
import avicit.bdp.dds.dispatch.model.TaskNode;
import avicit.bdp.dds.dispatch.model.TwoTuple;
import avicit.bdp.dds.dispatch.task.TaskTimeoutParameter;
import avicit.bdp.dds.common.Constants;
import avicit.bdp.dds.common.thread.Stopper;
import avicit.bdp.dds.dao.entity.TaskInstance;
import avicit.bdp.dds.remote.command.TaskKillRequestCommand;
import avicit.bdp.dds.remote.utils.Host;
import avicit.bdp.dds.server.master.cache.TaskInstanceCacheManager;
import avicit.bdp.dds.server.master.cache.impl.TaskInstanceCacheManagerImpl;
import avicit.bdp.dds.server.master.dispatch.context.ExecutionContext;
import avicit.bdp.dds.server.master.dispatch.enums.ExecutorType;
import avicit.bdp.dds.server.master.dispatch.executor.NettyExecutorManager;
import avicit.bdp.dds.server.registry.ZookeeperRegistryCenter;
import com.alibaba.fastjson2.JSON;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.commons.lang3.StringUtils;

import java.util.Date;
import java.util.Set;


/**
 * @author admin
 * master task exec thread
 */
public class MasterTaskExecThread extends MasterBaseTaskExecThread {

    /**
     * taskInstance state manager
     */
    private TaskInstanceCacheManager taskInstanceCacheManager;


    private NettyExecutorManager nettyExecutorManager;


    /**
     * zookeeper register center
     */
    private ZookeeperRegistryCenter zookeeperRegistryCenter;

    /**
     * constructor of MasterTaskExecThread
     *
     * @param taskInstance task instance
     */
    public MasterTaskExecThread(TaskInstance taskInstance) {
        super(taskInstance);
        this.taskInstanceCacheManager = SpringApplicationContext.getBean(TaskInstanceCacheManagerImpl.class);
        this.nettyExecutorManager = SpringApplicationContext.getBean(NettyExecutorManager.class);
        this.zookeeperRegistryCenter = SpringApplicationContext.getBean(ZookeeperRegistryCenter.class);
    }

    /**
     * get task instance
     *
     * @return TaskInstance
     */
    @Override
    public TaskInstance getTaskInstance() {
        return this.taskInstance;
    }

    /**
     * whether already Killed,default false
     */
    private boolean alreadyKilled = false;

    /**
     * submit task instance and wait complete
     *
     * @return true is task quit is true
     */
    @Override
    public TwoTuple<Boolean, String> submitWaitComplete() {
        TwoTuple<Boolean, String> result = new TwoTuple<>(false, null);
        this.taskInstance = submit();
        if (this.taskInstance == null) {
            logger.error("submit task instance to mysql and queue failed , please check and fix it");
            return result;
        }
        if (!ExecutionStatus.of(this.taskInstance.getState()).typeIsFinished()) {
            result = waitTaskQuit();
        }
        taskInstance.setEndTime(new Date());
        processService.updateTaskInstance(taskInstance);
        logger.info("task :{} id:{}, process id:{}, exec thread completed ",
                this.taskInstance.getName(), taskInstance.getId(), processInstance.getId());
        return result;
    }

    /**
     * polling db
     * <p>
     * wait task quit
     *
     * @return true if task quit success
     */
    public TwoTuple<Boolean, String> waitTaskQuit() {
        TwoTuple<Boolean, String> result = new TwoTuple<>(false, null);
        // query new state
        taskInstance = processService.findTaskInstanceById(taskInstance.getId());
        logger.info("wait task: process id: {}, task id:{}, task name:{} complete",
                this.taskInstance.getProcessInstanceId(), this.taskInstance.getId(), this.taskInstance.getName());
        // task time out
        boolean checkTimeout = false;
        TaskTimeoutParameter taskTimeoutParameter = getTaskTimeoutParameter();
        if (taskTimeoutParameter.getEnable()) {
            TaskTimeoutStrategy strategy = taskTimeoutParameter.getStrategy();
            if (strategy == TaskTimeoutStrategy.WARN || strategy == TaskTimeoutStrategy.WARNFAILED) {
                checkTimeout = true;
            }
        }

        while (Stopper.isRunning()) {
            try {
                if (this.processInstance == null) {
                    logger.error("process instance not exists , master task exec thread exit");
                    result.setFirst(true);
                    return result;
                }
                // task instance add queue , waiting worker to kill
                if (this.cancel || this.processInstance.getState() == ExecutionStatus.READY_STOP.getCode()) {
                    cancelTaskInstance();
                }
                if (processInstance.getState() == ExecutionStatus.READY_PAUSE.getCode()) {
                    pauseTask();
                }
                // task instance finished
                if (ExecutionStatus.of(taskInstance.getState()).typeIsFinished()) {
                    if (taskInstance.getState() == ExecutionStatus.SUCCESS.getCode()) {
                        TaskInstance cacheBean = taskInstanceCacheManager.getByTaskInstanceId(taskInstance.getId());
                        if (cacheBean != null && StringUtils.isNotEmpty(cacheBean.getResult())) {
                            result.setSecond(cacheBean.getResult());
                        }
                    }
                    // if task is final result , then remove taskInstance from cache
                    taskInstanceCacheManager.removeByTaskInstanceId(taskInstance.getId());
                    break;
                }
                if (checkTimeout) {
                    long remainTime = getRemaintime(taskTimeoutParameter.getInterval() * 60L);
                    if (remainTime < 0) {
                        logger.warn("task id: {} execution time out", taskInstance.getId());
                        // process define
                        //ProcessDefinition processDefine =
                        // processService.findProcessDefineById(processInstance.getProcessDefinitionId());
                        // send warn mail
                        alertDao.sendTaskTimeoutAlert(processInstance.getWarningUserIds(), processInstance.getId(),
                                processInstance.getName(), taskInstance.getId(), taskInstance.getName());
                        checkTimeout = false;
                    }
                }
                // updateProcessInstance task instance
                taskInstance = processService.findTaskInstanceById(taskInstance.getId());
                processInstance = processService.findProcessInstanceById(processInstance.getId());
                Thread.sleep(Constants.SLEEP_TIME_MILLIS);
            } catch (Exception e) {
                logger.error("exception", e);
                if (processInstance != null) {
                    logger.error("wait task quit failed, instance id:{}, task id:{}",
                            processInstance.getId(), taskInstance.getId());
                }
            }
        }
        result.setFirst(true);
        return result;
    }

    /**
     * pause task if task have not been dispatched to worker, do not dispatch anymore.
     */
    public void pauseTask() {
        taskInstance = processService.findTaskInstanceById(taskInstance.getId());
        if (taskInstance == null) {
            return;
        }
        if (StringUtils.isBlank(taskInstance.getHost())) {
            taskInstance.setState(ExecutionStatus.PAUSE.getCode());
            taskInstance.setEndTime(new Date());
            processService.updateTaskInstance(taskInstance);
        }
    }


    /**
     * task instance add queue , waiting worker to kill
     */
    private void cancelTaskInstance() throws Exception {
        if (alreadyKilled) {
            return;
        }
        alreadyKilled = true;

        taskInstance = processService.findTaskInstanceById(taskInstance.getId());

        if (StringUtils.isBlank(taskInstance.getHost())) {
            taskInstance.setState(ExecutionStatus.KILL.getCode());
            taskInstance.setEndTime(new Date());
            processService.updateTaskInstance(taskInstance);
            return;
        }

        TaskKillRequestCommand killCommand = new TaskKillRequestCommand();
        killCommand.setTaskInstanceId(taskInstance.getId());

        ExecutionContext executionContext = new ExecutionContext(killCommand.convert2Command(), ExecutorType.WORKER);

        Host host = Host.of(taskInstance.getHost());
        executionContext.setHost(host);

        nettyExecutorManager.executeDirectly(executionContext);

        logger.info("master kill taskInstance name :{} taskInstance id:{}, host: {}",
                taskInstance.getName(), taskInstance.getId(), taskInstance.getHost());
    }

    /**
     * whether exists valid worker group
     *
     * @param taskInstanceWorkerGroup taskInstanceWorkerGroup
     * @return whether exists
     */
    public Boolean existsValidWorkerGroup(String taskInstanceWorkerGroup) {
        Set<String> workerGroups = zookeeperRegistryCenter.getWorkerGroupDirectly();
        // not worker group
        if (CollectionUtils.isEmpty(workerGroups)) {
            return false;
        }

        // has worker group , but not taskInstance assigned worker group
        if (!workerGroups.contains(taskInstanceWorkerGroup)) {
            return false;
        }
        Set<String> workers = zookeeperRegistryCenter.getWorkerGroupNodesDirectly(taskInstanceWorkerGroup);
        if (CollectionUtils.isEmpty(workers)) {
            return false;
        }
        return true;
    }

    /**
     * get task timeout parameter
     *
     * @return TaskTimeoutParameter
     */
    private TaskTimeoutParameter getTaskTimeoutParameter() {
        String taskJson = taskInstance.getTaskJson();
        TaskNode taskNode = JSON.parseObject(taskJson, TaskNode.class);
        return taskNode.getTaskTimeoutParameter();
    }


    /**
     * get remain time?s?
     *
     * @return remain time
     */
    private long getRemaintime(long timeoutSeconds) {
        Date startTime = taskInstance.getStartTime();
        long usedTime = (System.currentTimeMillis() - startTime.getTime()) / 1000;
        return timeoutSeconds - usedTime;
    }
}
